package com.grf.hadoop.output_format;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class FilterRecordWriter extends RecordWriter<Text, NullWritable> {

    private FSDataOutputStream fosbaidu;
    private FSDataOutputStream fosother;

    public FilterRecordWriter(TaskAttemptContext job){
        try {
            FileSystem fs = FileSystem.get(job.getConfiguration());
            fosbaidu = fs.create(new Path("C:\\Users\\Administrator\\Desktop\\baidu.log"));
            fosother = fs.create(new Path("C:\\Users\\Administrator\\Desktop\\other.log"));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void write(Text key, NullWritable value) throws IOException,InterruptedException {
        if (key.toString().contains("baidu")) {
            fosbaidu.write(key.toString().getBytes());
        } else {
            fosother.write(key.toString().getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        IOUtils.closeStream(fosbaidu);
        IOUtils.closeStream(fosother);
    }
}
